package com.jz.pay.config;

import com.jz.pay.consumer.TopicType;
import com.jz.pay.consumer.topicListener.LogListerner;
import com.jz.pay.consumer.topicListener.RetryListener;
import com.jz.pay.consumer.topicListener.UserListerner;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RocketMQConsumerConfig {

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Value("${rocketmq.nameServer}")
    private String namesrvAddr;

    @Bean
    public DefaultMQPushConsumer logTopicConsumer(LogListerner logListerner) throws Exception {
        TopicType topicType = TopicType.LOG;
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup+topicType.getGroup());
        // 指定 Namesrv 地址
        consumer.setNamesrvAddr(namesrvAddr);
        // 设置消费起点
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setPullBatchSize(2);
        consumer.setConsumeMessageBatchMaxSize(2);
        // 设置每个 Topic 的监听器
        consumer.subscribe(topicType.getValue(), "*");
        consumer.setMessageListener(logListerner);
        // 启动消费者实例
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer logRetryTopicConsumer(RetryListener retryListener) throws Exception {
        TopicType topicType = TopicType.LOG;
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup+topicType.getRetryGroup());
        // 指定 Namesrv 地址
        consumer.setNamesrvAddr(namesrvAddr);
        // 设置消费起点
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setPullBatchSize(2);
        consumer.setConsumeMessageBatchMaxSize(2);
        // 设置每个 Topic 的监听器
        consumer.subscribe(topicType.getRetry(), "*");
        consumer.setMessageListener(retryListener);
        // 启动消费者实例
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer userTopicConsumer(UserListerner userListerner) throws Exception {
        TopicType topicType = TopicType.USER;
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup+topicType.getValue());
        // 指定 Namesrv 地址
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setPullBatchSize(2);
        consumer.setConsumeMessageBatchMaxSize(2);
        // 设置消费起点
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置每个 Topic 的监听器
        consumer.subscribe(topicType.getValue(), "*");
        consumer.setMessageListener(userListerner);
        // 启动消费者实例
        consumer.start();
        return consumer;
    }
}
